package com.company.wushuo.justfortest;

import android.support.annotation.NonNull;
import android.util.Log;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;


/**
 * Created by wushuo on 2021/4/6
 */
public class RxJavaTest {

    private static Executor executor = Executors.newFixedThreadPool(5);

    static void test() {
        Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                subscriber.onNext(10);
                subscriber.onNext(9);
                subscriber.onCompleted();
            }
        })
//                .map(new Func1<Integer, String>() {
//
//            @Override
//            public String call(Integer integer) {
//                Log.d("wushuo", "");
//                return integer + " from convert";
//            }
//        })
                .observeOn(Schedulers.io())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.d("wushuo", "received onComplete");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d("wushuo", "received " + integer);
                    }
                });
    }

    public static void test2() {
        // D/Wushuo: 【pool-1-thread-1 】  19
        // 【pool-1-thread-2 】  20
        // 【pool-1-thread-3 】  21
        // 【pool-1-thread-4 】  22
        // 【pool-1-thread-5 】  23
        // 【pool-1-thread-1 】  24
        // 【pool-1-thread-2 】  25
        // 【pool-1-thread-3 】  26
        // 可见observe不一定是在同一个线程中
        doSomethingInExecutor();
        Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
//                for (int i = 0; i < 30; i++) {
//                    try {
//                        Thread.sleep(10);
//                    } catch (InterruptedException e) {
//                        e.printStackTrace();
//                    }
//                    subscriber.onNext(i);
//                }
                subscriber.onNext(1);
                subscriber.onCompleted();
                subscriber.onNext(2);
            }
        })
                .observeOn(namedScheduler("1 - subscribeOn"))
//                .doOnSubscribe(new Action0() {
//                    @Override
//                    public void call() {
//                        log("1 - doOnSubscribe");
//                    }
//                })
//                .subscribeOn(namedScheduler("2 - subscribeOn"))
//                .doOnSubscribe(new Action0() {
//                    @Override
//                    public void call() {
//                        log("2 - doOnSubscribe");
//                    }
//                })
//                .doOnNext(new Action1<Integer>() {
//                    @Override
//                    public void call(Integer integer) {
//                        log("onNext");
//                    }
//                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        log(" " + integer);
                    }
                });
    }

    private static void doSomethingInExecutor() {
        for (int i = 0; i < 10; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    private static void log(String log) {
        Log.d("Wushuo", "【" + Thread.currentThread().getName() + " 】 " + log);
    }

    private static Scheduler namedScheduler(final String name) {
//        return Schedulers.from(executor);
        return new NamedThreadSchedulers(new ThreadFactory() {
            @Override
            public Thread newThread(@NonNull Runnable r) {
                return new Thread(r, name);
            }
        });
//        return Schedulers.io();
    }

    public static void testSubscriberOn() {
        new Thread("my-test-thread") {
            @Override
            public void run() {
                Observable.create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        Log.d("Wushuo", "emit name: " + Thread.currentThread().getName());
                        subscriber.onNext(1);
                        subscriber.onCompleted();
                    }
                })
                        .map(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer) {
                        Log.d("Wushuo", "map1: thread name: " + Thread.currentThread().getName() + "  " + integer);
                        return integer + 1;
                    }
                })
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribeOn(Schedulers.io())
                        .map(new Func1<Integer, Integer>() {
                            @Override
                            public Integer call(Integer integer) {
                                Log.d("Wushuo", "thread name: " + Thread.currentThread().getName() + "  " + integer);
                                return integer * 10;
                            }
                        })
                        .subscribeOn(namedScheduler("my-schedular"))
                        .subscribe(new Subscriber<Integer>() {
                            @Override
                            public void onCompleted() {
                                Log.d("Wushuo", "observer thread name complete: " + Thread.currentThread().getName());
                            }

                            @Override
                            public void onError(Throwable e) {

                            }

                            @Override
                            public void onNext(Integer s) {
                                Log.d("Wushuo", "observer thread name: " + Thread.currentThread().getName());
                            }
                        });
            }
        }.start();
    }

    //对象类
    static class Data {
        public String name;
        public Data(String name) {
            this.name = name;
        }
    }

    // 测试subscribe两次
    public static void testSubcribeTwice() {
        //比如我们使用just操作符来发送二个Data对象
        Observable<Data> data = Observable.just(
                new Data("aaaa"),
                new Data( "bbbb")
        );

//        Observable<Data> data = Observable.create(new Observable.OnSubscribe<Data>() {
//            @Override
//            public void call(Subscriber<? super Data> subscriber) {
//                Log.d("Wushuo", "subscriber is " + subscriber);
//                subscriber.onNext(new Data("aaaa"));
//                subscriber.onNext(new Data("bbbb"));
//                subscriber.onCompleted();
//            }
//        });

        //比如我们用一个Consumer对象来订阅
        data.subscribe(new Action1<Data>() {
            @Override
            public void call(Data data) {
                //里面的内容是把Observable发送过来的对象里面的name属性值改成cccc
                data.name = "cccc";
                Log.d("Wushuo", "first changed to: " + data.name);
            }
        });

        //这时候我们在用一个新的Consumer来订阅这个Observable
        data.subscribe(new Action1<Data>() {
            @Override
            public void call(Data data) {
                //这时候打印data.name
                // 但是对于just创建的observable，打印的结果是
                // cccc、 cccc
                // 对于create创建的observable，打印的结果是
                // aaaa、bbbb
                // 这是因为多个subscribe实际上调用了多次Observable.OnSubscribe.call方法
                // 也就是                Log.d("Wushuo", "subscriber is " + subscriber);
                // 这个log被打印了两次

                // create完整的log:
//                subscriber is rx.observers.SafeSubscriber@d587227
//                first changed to: cccc
//                first changed to: cccc
//                subscriber is rx.observers.SafeSubscriber@29aacd4
//                aaaa
//                bbbb
                // 所以核心原因还是just已经生成了observable
                Log.d("Wushuo", data.name);
            }
        });
    }
}
